package org.luxor.commons.websocket.component.session;

import org.luxor.commons.redis.client.RedisTopicClient;
import org.luxor.commons.websocket.component.protocol.RemoveCommand;
import org.luxor.commons.websocket.component.protocol.SendAllCommand;
import org.luxor.commons.websocket.component.protocol.SendCommand;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.websocket.RemoteEndpoint;
import javax.websocket.Session;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * WebSocket会话共享存储器
 *
 * @author Mr.yan  @date 2021/9/4
 */
public class WsShareSessionStorage implements WsSessionStorage {
    private static final Logger logger = LoggerFactory.getLogger(WsSessionStorage.class);

    /**
     * 会话内存存储器
     */
    private final ConcurrentMap<String, Session> memoryStore = new ConcurrentHashMap();

    /**
     * 已经处理消息的HASH记录
     */
    private List<Long> processedRecord = new ArrayList<>();

    private RedisTopicClient redisTopicClient;

    public WsShareSessionStorage(RedisTopicClient redisTopicClient) {
        this.redisTopicClient = redisTopicClient;
    }

    /**
     * 添加会话
     *
     * @param token
     * @param session
     */
    @Override
    public void addSession(String token, Session session) {
        if (existSession(token)) {
            logger.info("cover the existing token-session. token:" + token);
        }
        memoryStore.put(token, session);
    }

    /**
     * 移除会话
     *
     * @param token
     */
    @Override
    public void removeSession(String token) {
        if (existSession(token)) {
            memoryStore.remove(token);
        } else {
            redisTopicClient.publish(new RemoveCommand(token).toJson(), WsShareProtocolReceiver.WS_TOPIC);
        }
    }

    /**
     * 发送消息
     *
     * @param token
     * @param message
     * @return
     */
    @Override
    public void send(String token, String message) {
        if (existSession(token)) {
            RemoteEndpoint.Basic remoteEndpoint = memoryStore.get(token).getBasicRemote();
            try {
                remoteEndpoint.sendText(message);
            } catch (IOException e) {
                logger.warn("fail to send, token:" + token, e);
            }
        } else {
            this.redisTopicClient.publish(new SendCommand(token, message).toJson(), WsShareProtocolReceiver.WS_TOPIC);
        }
    }

    /**
     * 群发消息
     *
     * @return
     */
    @Override
    public void sendAll(String message) {
        processedRecord.add(Long.valueOf(message.hashCode()));

        for (Map.Entry<String, Session> entry : memoryStore.entrySet()) {
            try {
                RemoteEndpoint.Basic remoteEndpoint = entry.getValue().getBasicRemote();
                remoteEndpoint.sendText(message);
            } catch (IOException e) {
                logger.warn("fail to send, token:" + entry.getKey(), e);
            }
        }
        redisTopicClient.publish(new SendAllCommand(message).toJson(), WsShareProtocolReceiver.WS_TOPIC);
    }

    /**
     * 存在会话？
     *
     * @param token
     * @return
     */
    @Override
    public boolean existSession(String token) {
        return memoryStore.keySet().contains(token);
    }


    /**
     * 单纯移除会话
     *
     * @param token
     */
    public void pureRemove(String token) {
        if (existSession(token)) {
            memoryStore.remove(token);
        }
    }

    /**
     * 普通消息
     *
     * @param token
     * @param message
     * @return
     */
    public void pureSend(String token, String message) {
        if (existSession(token)) {
            RemoteEndpoint.Basic remoteEndpoint = memoryStore.get(token).getBasicRemote();
            try {
                remoteEndpoint.sendText(message);
            } catch (IOException e) {
                logger.warn("fail to send, token:" + token, e);
            }
        }
    }

    /**
     * 群发消息
     *
     * @param message
     * @return void
     */
    public void pureSendAll(String message) {
        Iterator<Long> it = processedRecord.iterator();
        while (it.hasNext()) {
            Long pr = it.next();
            if (pr.longValue() == message.hashCode()) {
                processedRecord.remove(pr);
                return;
            }
        }

        for (Map.Entry<String, Session> entry : memoryStore.entrySet()) {
            try {
                RemoteEndpoint.Basic remoteEndpoint = entry.getValue().getBasicRemote();
                remoteEndpoint.sendText(message);
            } catch (IOException e) {
                logger.warn("fail to send, token:" + entry.getKey(), e);
            }
        }
    }

}
